package com.wangjiacai.utils.ali;


import com.alibaba.nls.client.transport.ConnectionListener;

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;

public class AliPool {

    protected NettyNlsClient nettyNlsClient;
    protected int coreSize;
    protected int maxSize;
    protected AtomicInteger waitNum;
    protected AtomicInteger workSize;
    protected Queue<NettyConnection> free;
    protected Queue<NettyConnection> work;
    protected PingThread pingThread;

    public AliPool(NettyNlsClient nettyNlsClient, int maxSize) {
        this.nettyNlsClient=nettyNlsClient;
        this.maxSize = Math.max(coreSize, maxSize);
        this.waitNum=new AtomicInteger(0);
        this.workSize=new AtomicInteger(0);
    }

    public AliPool(NettyNlsClient nettyNlsClient, int coreSize, int maxSize) {
        this.nettyNlsClient=nettyNlsClient;
        this.coreSize = coreSize;
        this.maxSize = Math.max(coreSize, maxSize);
        this.waitNum=new AtomicInteger(0);
        this.free=new ArrayDeque<>();
        this.work=new ArrayDeque<>();
        this.pingThread=new PingThread();
        this.pingThread.setDaemon(true);
        this.pingThread.start();
    }

    private NettyConnection getFreeConnection(ConnectionListener connectionListener){
        NettyConnection nettyConnection = free.poll();
        if (nettyConnection != null) {
            nettyConnection.getNettySpeechTranscriberListener().setConnectionListener(connectionListener);
        }
        return nettyConnection;
    }

    public NettyConnection getConnection(NettySpeechListener listener, ConnectionListener connectionListener) throws Exception {
        NettyConnection nettyConnection=getFreeConnection(connectionListener);
        if(nettyConnection!=null) return nettyConnection;
        while (work.size() >= maxSize) {
            synchronized (work) {
                waitNum.getAndIncrement();
                work.wait();
            }
        }
        waitNum.getAndDecrement();
        nettyConnection=getFreeConnection(connectionListener);
        if(nettyConnection!=null) return nettyConnection;
        nettyConnection = nettyNlsClient.connect(listener);
        nettyConnection.setAliPool(this);
        listener.setConnectionListener(connectionListener);
        work.add(nettyConnection);
        return nettyConnection;
    }

    public void free(NettyConnection nettyConnection){
        if(work.size()>=coreSize){
            work.remove(nettyConnection);
            nettyConnection.getChannel().close();
        }else free.add(nettyConnection);
        synchronized (work){
            work.notify();
        }
    }

    class PingThread extends Thread{
        public PingThread() {
            super("ali asr ping thread");
        }
        @Override
        public void run() {
            while (true){
                synchronized (work) {
                    work.stream().forEach(nettyConnection -> {
                        if(!nettyConnection.isActive()){
                            work.remove(nettyConnection);
                            nettyConnection.getChannel().close();
                            work.notify();
                            return;
                        }
                        if (nettyConnection.isActive()&&work.size() >= coreSize) {
                            if(waitNum.get()!=0) free.add(nettyConnection);
                            else {
                                work.remove(nettyConnection);
                                nettyConnection.getChannel().close();
                            }
                            work.notify();
                            return;
                        }
                        nettyConnection.sendPing();
                    });
                }
                try {
                    Thread.sleep(1000*10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
